package com.aioething.iot.mqtt;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.nio.charset.StandardCharsets;

@Slf4j
public class VirtualClient implements Runnable {
    private final static String USER_NAME = "100001";
    private final static String CLIENT_ID = "869455060089535";


    private String clientId;

    public VirtualClient(String clientId) {
        this.clientId = clientId;
    }
    @SneakyThrows
    @Override
    public void run() {
        MqttClient client = new MqttClient("tcp://127.0.0.1:1883", clientId, new MemoryPersistence());
        MqttConnectOptions options = new MqttConnectOptions();
        //是否清空session，设置false表示服务器会保留客户端的连接记录（订阅主题，qos）,客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
        //设置为true表示每次连接服务器都是以新的身份
        options.setCleanSession(true);
        //设置连接用户名
        options.setUserName(USER_NAME);
        //设置连接密码
        options.setPassword("123456".toCharArray());
        //设置超时时间，单位为秒
        options.setConnectionTimeout(100);
        //设置心跳时间 单位为秒，表示服务器每隔 1.5*20秒的时间向客户端发送心跳判断客户端是否在线
        options.setKeepAliveInterval(60);
        //设置回调
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {
                run();
            }

            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                String items[] = topic.split("/");
                JSONObject payload = JSONUtil.parseObj(new String(mqttMessage.getPayload()));
                log.info(String.format("topic: %s payload: %s", topic, new String(mqttMessage.getPayload())));
                if ("down".equals(items[3]) && "ota".equals(items[4]) && items.length == 5) {
                    new Thread(() -> {
                        MqttMessage resMessage = new MqttMessage();
                        resMessage.setQos(0);
                        resMessage.setRetained(false);
                        JSONObject params = JSONUtil.createObj().set("method", "ota_reply").set("clientToken", payload.getStr("clientToken")).set("code", 0);
                        resMessage.setPayload(params.toString().getBytes(StandardCharsets.UTF_8));
                        String resTopic = String.format("things/%s/%s/up/ota", items[1], items[2]);
                        MqttTopic mqttTopic = client.getTopic(resTopic);
                        MqttDeliveryToken token;
                        try {
                            token = mqttTopic.publish(resMessage);
                            token.waitForCompletion();
                        } catch (MqttException e) {
                            e.printStackTrace();
                        }
                    }).start();
                }
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

            }
        });
        client.connect(options);
        if (client.isConnected()) {
            client.subscribe(String.format("things/%s/%s/down/property", USER_NAME, CLIENT_ID));
            client.subscribe(String.format("things/%s/%s/down/action/+", USER_NAME, CLIENT_ID));

        }
    }

    public static void main(String[] args) throws MqttException, InterruptedException {
        new Thread(new VirtualClient("869455060089535")).start();
        while (true) {
            Thread.sleep(1000);
        }
    }
}
